package com.cskaoyan.order.mq.delay;

import com.cskaoyan.mall.order.constant.OrderRetCode;
import com.cskaoyan.order.constant.OrderConstants;
import com.cskaoyan.order.dal.entitys.Order;
import com.cskaoyan.order.dal.persistence.OrderMapper;
import com.cskaoyan.order.dto.CancelOrderRequest;
import com.cskaoyan.order.dto.CancelOrderResponse;
import com.cskaoyan.order.service.OrderCoreService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * @author LY
 * @since 2022/06/21 21:11
 */
@Slf4j
@Component
public class DelayOrderCancelConsumer {

    private DefaultMQPushConsumer consumer;

    @Autowired
    OrderMapper orderMapper;

    @Autowired
    OrderCoreService orderCoreService;

    @PostConstruct
    public void init(){
        // 1. 创建consumer对象
        consumer = new DefaultMQPushConsumer("delay_order_cancel_consumer");
        // 2. 设置nameserver地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        try {
            // 3. 订阅主题Topic
            consumer.subscribe("delay_order_cancel", "*");
            // 4. 设置消息监听器
            consumer.setMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    // 1. 获取OrderId
                    MessageExt message = list.get(0);
                    byte[] data = message.getBody();
                    try {
                        String orderId = new String(data, 0, data.length, "utf-8");
                        // 2. 获取订单状态，若已支付或者已取消，什么都不做；若查询不到订单，说明已删除，同样什么都不做
                        Order order = orderMapper.selectByPrimaryKey(orderId);
                        if(order == null){
                            // 订单已删除
                            log.info("order had been delete {}", orderId);
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        Integer status = order.getStatus();
                        if(status == OrderConstants.ORDER_STATUS_PAYED || status == OrderConstants.ORDER_STATUS_TRANSACTION_CANCEL){
                            log.info("order had been paid or cancelled {}", orderId);
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        // 3. 取消订单
                        CancelOrderRequest request = new CancelOrderRequest();
                        request.setOrderId(orderId);
                        CancelOrderResponse response = orderCoreService.cancelOrder(request);
                        if(OrderRetCode.SUCCESS.getCode().equals(response.getCode())){
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });
            // 5. 启动consumer
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
